package cn.taskservice.jwt.serviceImpl;

import cn.taskservice.jwt.config.JwtPropertiesConfig;
import cn.taskservice.jwt.service.TokenStoreService;
import com.github.benmanes.caffeine.cache.Cache;
import com.github.benmanes.caffeine.cache.Caffeine;
import jakarta.annotation.PostConstruct;
import jakarta.annotation.PreDestroy;
import lombok.RequiredArgsConstructor;
import lombok.extern.slf4j.Slf4j;

import org.springframework.data.redis.connection.Message;
import org.springframework.data.redis.connection.MessageListener;
import org.springframework.data.redis.core.StringRedisTemplate;
import org.springframework.data.redis.listener.ChannelTopic;
import org.springframework.data.redis.listener.RedisMessageListenerContainer;



import java.nio.charset.StandardCharsets;
import java.util.concurrent.TimeUnit;

/**
 * Caffeine+ redis 缓存 token,分布式部署时，多节点使用redis 发布/订阅功能实现多节点本地缓存同步
 */
@Slf4j
@RequiredArgsConstructor
public class LocalCacheTokenStoreServiceImpl implements TokenStoreService, MessageListener {

    private static final String TOKEN_RENEW_KEY = "jwt:renew:";
    private static final String TOKEN_BLACKLIST_KEY = "jwt:blacklist:";
    private static final String REDIS_CHANNEL = "jwt:blacklist:channel";

    private final StringRedisTemplate redisTemplate;
    private final JwtPropertiesConfig jwtProperties;
    private final RedisMessageListenerContainer redisMessageListenerContainer;

    private Cache<String, Boolean> localBlacklistCache;
    private ChannelTopic topic;

    @PostConstruct
    public void init() {
        // 初始化本地缓存，过期时间和JWT过期时间保持一致
        this.localBlacklistCache = Caffeine.newBuilder()
                .maximumSize(10_000)
                .expireAfterWrite(jwtProperties.getExpiration(), TimeUnit.SECONDS)
                .build();

        // 订阅频道
        topic = new ChannelTopic(REDIS_CHANNEL);
        redisMessageListenerContainer.addMessageListener(this, topic);
        log.info("Subscribed to Redis channel: {}", REDIS_CHANNEL);
    }

    @PreDestroy
    public void cleanup() {
        if (topic != null) {
            redisMessageListenerContainer.removeMessageListener(this, topic);
            log.info("Unsubscribed from Redis channel: {}", REDIS_CHANNEL);
        }
    }

    @Override
    public boolean shouldRenew(String token) {
        String key = TOKEN_RENEW_KEY + token;
        Boolean hasKey = redisTemplate.hasKey(key);
        return hasKey == null || !hasKey;
    }

    @Override
    public void markAsRenewed(String token) {
        String key = TOKEN_RENEW_KEY + token;
        redisTemplate.opsForValue().set(key, "1", jwtProperties.getRenewInterval(), TimeUnit.SECONDS);
    }

    @Override
    public void invalidateToken(String token) {
        String key = TOKEN_BLACKLIST_KEY + token;
        // Redis 设置黑名单
        redisTemplate.opsForValue().set(key, "1", jwtProperties.getExpiration(), TimeUnit.SECONDS);
        // 本地缓存更新
        localBlacklistCache.put(token, Boolean.TRUE);
        // 发布消息通知其它实例同步
        redisTemplate.convertAndSend(REDIS_CHANNEL, token);
        log.info("Token invalidated and published to channel: {}", token);
    }

    @Override
    public boolean isBlacklisted(String token) {
        Boolean cached = localBlacklistCache.getIfPresent(token);
        if (cached != null && cached) {
            return true;
        }
        String key = TOKEN_BLACKLIST_KEY + token;
        Boolean redisHasKey = redisTemplate.hasKey(key);
        if (redisHasKey != null && redisHasKey) {
            localBlacklistCache.put(token, Boolean.TRUE);
            return true;
        }
        return false;
    }

    /**
     * Redis消息监听回调
     * 当接收到黑名单token消息时，更新本地缓存
     */
    @Override
    public void onMessage(Message message, byte[] pattern) {
        String token = new String(message.getBody(), StandardCharsets.UTF_8);
        log.info("Received Redis pub/sub message for token blacklist sync: {}", token);
        localBlacklistCache.put(token, Boolean.TRUE);
    }
}
